package cn.huq.day03

import org.apache.spark.rdd.{RDD, ShuffledRDD}
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}


object NoShuffleDemo {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("FlatMapValuesDemo").setMaster("local")
    val sc = new SparkContext(conf)

    val arr: Array[(String, Int)] = Array(
      ("scala", 5), ("hadoop", 8),("scala", 2), ("hadoop", 3),
      ("hadoop", 1),("flink", 2), ("scala", 5),("scala", 2), ("hadoop", 3))
    val rdd: RDD[(String, Int)] = sc.parallelize(arr, 2)

    val rdd1: RDD[(String, Int)] = rdd.partitionBy(new HashPartitioner(2))
    val rdd3: ShuffledRDD[String, Int, Int] =
      new ShuffledRDD[String, Int, Int](rdd, new HashPartitioner(rdd.partitions.length))
    val rdd2: RDD[(String, Int)] = rdd1.reduceByKey(new HashPartitioner(3), _ + _)
    rdd2.saveAsTextFile("datas/output2")

    Thread.sleep(1000000)
  }

}
